package com.zck2

import com.zck2.utils.UDFPartitioner
import org.apache.spark.rdd.RDD

import scala.collection.mutable
import scala.collection.mutable.HashSet

object AggAndFilter {

//  def aggFilterAB(rddA: RDD[(String, Array[String])], rddB: RDD[(String, Array[String])]): RDD[(String, Array[String])] = {
//    val rddAB = rddA.union(rddB)
//
//    rddAB.groupBy(_._1)
//      .mapValues { iter =>
//        val set = new HashSet[String]()
//        iter.foreach { case (_, values) => values.foreach(set.add) }
//        set.toArray
//      }
//      .map { case (k, v) => (k, v.distinct) } // 再次去重，以防万一
//  }
  def aggFilterAB(rddA: RDD[(String, Array[String])], rddB: RDD[(String, Array[String])]): RDD[(String, Array[String])] = {
    val rddAB = rddA.union(rddB)

    rddAB.groupBy(_._1)
      .mapValues { iter =>
        val set = new HashSet[String]()
        iter.foreach { case (_, values) => values.foreach(set.add) }
        set.toArray  // 转换为去重后的数组
      }
  }
  def isSubArray(subArray: Array[String], array: Array[String]): Boolean = {
    subArray.forall(array.contains)
  }

  def delDistinct(rddRes: RDD[(String, Array[String])]): RDD[(String, Array[String])] = {
    val rdd_init: RDD[(String, (String, Array[String]))] = rddRes.map(x => {
      (x._1.split("_")(0), (x._1, x._2))
    })

    val arr_k = rdd_init.map(_._1.toLong).distinct().collect()
    val paredRdd = rdd_init.partitionBy(new UDFPartitioner(arr_k))

    paredRdd.mapPartitions(iter => {
      val rows = iter.toArray.sortBy(_._2._2.length).reverse
      val buffer = mutable.ListBuffer.empty[(String, (String, Array[String]))]

      if (rows.length >= 2) {
        for (i <- 0 until rows.length - 1) {
          val (key1, (k1, v1)) = rows(i) // 第一行元素
          var v1Updated = mutable.ArrayBuffer.empty[String] ++ v1

          for (j <- i + 1 until rows.length) {
            val (key2, (k2, v2)) = rows(j) // 第二行元素

            if (isSubArray(v2, v1)) {
              if (!v1Updated.contains(k2)) {
                // 将 k2 添加到 v1 中
                v1Updated :+= k2
              }
              // 从 buffer 中移除 k2 所在的行
              buffer --= buffer.filter(_._2._1 == k2)

            }
          }
          // 更新 v1 的值为添加了子数组元素的新数组
          val updatedRow = (key1, (k1, v1Updated.toArray))
          buffer += updatedRow
        }
      }
      else {
        buffer ++= rows
      }
      buffer.iterator.map { case (key, (k, v)) => (k, v) }
    }
    )

  }

//  def getResFile(rddRes: RDD[(String, Array[String])], output: String): Unit = {
//    // 对每个 (key, value) 对，首先将相同 key 的 value 合并
//    val rddkv = rddRes.flatMap { case (k, v) =>
//        v.map(x => (k, x))  // 将每个数组 v 中的元素 x 展平成 (k, x) 形式
//      }
//      .distinct()  // 对 (k, v) 对进行去重
//
//    // 按 key 聚合相同 key 的 value，确保每个 key 只有一个对应的去重后的 value 列表
//    val aggregated = rddkv
//      .groupByKey()  // 按照 key 进行分组
//      .mapValues(_.toArray.distinct.mkString(","))  // 对每个 key 下的 value 数组去重并合并成一个逗号分隔的字符串
//
//    // 将结果按 key 排序，并转换成最终的字符串格式
//    aggregated
//      .sortByKey()  // 按 key 排序
//      .map { case (k, v) => s"$k,$v" }  // 转换成字符串格式 "key,value1,value2,..."
//      .coalesce(1)  // 将所有数据收集到一个分区
//      .saveAsTextFile(output)  // 保存到文件
//  }

  def getResFile(rddRes: RDD[(String, Array[String])], output: String): Unit = {
    // 先对每个数组 v 进行去重处理（不需要再在reduceByKey中去重）
    val rddkv = rddRes
      .flatMap { case (k, v) =>
        v.distinct.map(x => (k, x))  // 对每个数组 v 中的元素 x 进行去重后展开
      }
    // 使用 reduceByKey 来聚合相同 key 的 value，避免 groupByKey 的高开销
    val aggregated = rddkv
      .reduceByKey((a, b) => {
        // 使用 HashSet 来合并两个字符串中的元素，并进行去重
        val set = (a.split(",") ++ b.split(",")).toSet  // 合并两个字符串，去重
        set.mkString(",")  // 将去重后的元素重新拼接成逗号分隔的字符串
      })
    // 按 key 排序并转换成最终的字符串格式
    aggregated
      .sortByKey()  // 按 key 排序
      .map { case (k, v) => s"$k,$v" }  // 转换成字符串格式 "key,value1,value2,..."
      .coalesce(1)  // 将所有数据收集到一个分区
      .saveAsTextFile(output)  // 保存到文件
  }

  //统计证认结果中的无重复星体数目
  def getNum(rddRes: RDD[(String, Array[String])]) = {
    val num = rddRes.repartition(1).map {
      case (k, v) =>
        val vv = v.mkString(",")
        s"$k,$vv"
    }.reduce(_ + "," + _).split(",").distinct.length

    println(s"星体数: $num")
    num
  }

  def getNum2(rddRes: RDD[(String, Array[String])]) = {
        val rddkv: RDD[(String, String)] = rddRes.map((x: (String, Array[String])) => {
          val k = x._1
          val v = x._2.distinct.mkString(",")
          (k, v)
        }).map(_.swap).groupByKey().flatMapValues(_.take(1)).map(_.swap) //去简单冗余 不同K相同V ==>只保留一行

        //    去简单冗余后的格式恢复
        val rdd: RDD[(String, Array[String])] = rddkv.map(x => {
          val k = x._1
          val v = x._2.split(",")
          (k, v)
        }
        )

    val eNum = rdd.map(_._2.length).reduce(_ + _)

    eNum
  }


}
